package net.bwie.realtime.jtp.dws.log.job;

import com.mysql.cj.result.Row;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.TableFunction;

/*
 * 搜索关键词
 */
public class JtpTrafficSearchKeywordMinuteWindowDwsJob {
    public static void main(String[] args) {


        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv() ;

        // 2. 映射到Kafka消息队列 ddl
        createInputTable(tabEnv);

        // 3. 数据处理-select
        Table reportTable = handle(tabEnv);

        // 4. 输出表-output：映射到Doris表 ddl
         createOutputTable(tabEnv) ;

        // 5. 保存数据-insert
        saveToSink(tabEnv, reportTable) ;

    }


    // 1.构建flink sql执行环境
    private static TableEnvironment getTableEnv() {
//        TableEnvironment tabEnv , Table resultTable

        EnvironmentSettings settings =  EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tabEv = TableEnvironment.create(settings);

        Configuration configuration = tabEv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");

        return tabEv;
    }

//    2.输入表-input：
    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE dws_traffic_search_keyword_window_report_doris_sink\n" +
                "(\n" +
                "    `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                "    `window_end_time`   STRING COMMENT '窗口结束日期时间',\n" +
                "    `keyword`           STRING COMMENT '搜索关键词',\n" +
                "    `keyword_count`     BIGINT COMMENT '搜索关键词被搜索次数',\n" +
                "    `ts`                BIGINT COMMENT '数据产生时间戳'\n" +
                ") WITH (\n" +
                "      'connector' = 'clickhouse',\n" +
                "      'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                "      'table' = 'dws_traffic_search_keyword_window_report',\n" +
                "      'username' = 'default',\n" +
                "      'password' = '',\n" +
                "      'format' = 'json'\n" +
                "      )"
        );
    }

    private static Table handle(TableEnvironment tabEnv) {
        // 获取搜索词 和 搜索时间
        Table searchLogTable = tabEnv.sqlQuery(
                "SELECT \n" +
                        "page['item'] as full_word \n" +
                        ",row_time \n" +
                        "FROM dwd_traffic_page_log_kafka_source \n" +
                        "WHERE page['item'] IS NOT NULL \n" +
                        " AND page['last_page_id'] = 'search' \n" +
                        " AND page['item_type'] = 'keyword'"
        );
        tabEnv.createTemporaryView("search_log_table", searchLogTable);

        // 使用自定义UDTF函数 对搜索词进行中文分词
        tabEnv.createTemporarySystemFunction("ik_analzer_udtf",IKAnalyzerFunction.class);


        return null;
    }

    @FunctionHint(output = @DataTypeHint("ROW<keyword STRING>"))
    private stat  c class IKAnalyzerFunction extends TableFunction<Row> {
        public void eval(String fullWord) throws Exception {
            AnalyzerUtil.
        }
    }



    private static void createOutputTable(TableEnvironment tabEnv) {

    }

    private static void saveToSink(TableEnvironment tabEnv, Table reportTable) {

    }
}
